package com.xzx.ecommercedw.hiveudf;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.hive.ql.exec.UDFArgumentException;
import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDTF;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorFactory;
import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorFactory;
import org.json.JSONArray;

import java.util.ArrayList;

/**
 * @author xinzhixuan
 * @version 1.0
 * @date 2024/9/24 17:42
 */
public class EventLogEventJsonUDTF extends GenericUDTF {

    @Override
    public StructObjectInspector initialize(StructObjectInspector argOIs) throws UDFArgumentException {
        ArrayList<String> fieldNames = new ArrayList<>();
        ArrayList<ObjectInspector> fieldOIs = new ArrayList<>();

        fieldNames.add("event_name");
        fieldOIs.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);
        fieldNames.add("event_json");
        fieldOIs.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);

        return ObjectInspectorFactory.getStandardStructObjectInspector(fieldNames, fieldOIs);
    }

    // 一进多出，所有只有一个参数
    @Override
    public void process(Object[] args) throws HiveException {
        String input = args[0].toString();
         // 如果传进来的数据为空，则直接返回，过滤掉该数据
        if (StringUtils.isBlank(input)) {
            return;
        } else {
            JSONArray jsonArray = new JSONArray(input);
            if (jsonArray.length() == 0) {
                return;
            }

            for (int i = 0; i < jsonArray.length(); i++) {
                //[{"ett":"1726838519609","en":"display","kv":{"action":"2","goodsid":"24","place":"4","extend1":"1","category":"77"}},]
                String eventName = jsonArray.getJSONObject(i).getString("en");
                String eventJson = jsonArray.getString(i);
                forward(new String[]{eventName, eventJson});
            }
        }
    }

    @Override
    public void close() throws HiveException {

    }
}
